package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.storage.StorageLevel;

import scala.Tuple2;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.google.common.collect.Lists;
import com.surfilter.massdata.spark.bean.DnsCount_jr_city;
import com.surfilter.massdata.spark.bean.SA7601;
import com.surfilter.massdata.spark.model.DnsStatConsts;
import com.surfilter.massdata.spark.model.StructIP;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.IPDatabase;

public class DomainAnalyzTask_SA7601JR_city_dayNew implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(DomainAnalyzTask_SA7601JR_city_dayNew.class);
	private static final long serialVersionUID = 1;
	private static String outName;
	
	
	@SuppressWarnings({ "resource" })
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		long temp=start;
		long temp_end=start;
		double exectime=0.0;
		
		System.out.println("==============DomainAnalyzTask_SA7601_city_dayNew begin====================");
		System.out.println("............本省接入(按地市)....开始............");
		
		List<OutQueueEntity> outlist = new ArrayList<OutQueueEntity>();
		JavaSparkContext sc ;
		try {
			DataFrame dataSource = dataFrames.get("day_domain");
			String sql = "select top_domain,ip,ip_visit_count from day_domain_temp ";
			DataFrame df_sa7601 = dataSource.sqlContext().sql(sql);
			
			System.out.println("............关联br2002表，过滤dip............");
			sc = new JavaSparkContext(dataSource.sqlContext().sparkContext());
			final Broadcast<IPDatabase> ipBroadcast = sc.broadcast(IPDatabase.getInstance());
			JavaRDD<String> conect = df_sa7601.toJavaRDD().flatMap(new FlatMapFunction<Row, String>() {
				private static final long serialVersionUID = 1L;
				@Override
				public Iterable<String> call(Row t) throws Exception {
					IPDatabase ipDatabase = ipBroadcast.value();
					List<String> resultList = Lists.newArrayList();
					String domain = t.getAs("top_domain").toString();
					String[] ip_str = t.getAs("ip").toString().split("\\|");
					String[] count_str = t.getAs("ip_visit_count").toString().split("\\|");
					String ip = null;
					String count = null;
					for(int i=0;i<=ip_str.length-1;i++){
						ip = ip_str[i];
						count = count_str[i];
						StructIP strIp = ipDatabase.get(ip);
						if(strIp !=null){
							String city = strIp.get(DnsStatConsts.IPExField.CITY);
							resultList.add(domain+"@"+city+"|"+count);
						}
					}
					return resultList;
				}
			});
			
			System.out.println("............切分数据,做reduceByKey............");
			JavaPairRDD<String, Long> visit_count = conect.mapToPair(new PairFunction<String, String, Long>() {
				private static final long serialVersionUID = 1L;
				@Override
				public Tuple2<String, Long> call(String t) throws Exception {
					try {
						String d_c= t.split("\\|")[0];
						if(StringUtils.isNotBlank(d_c)){
							long count = Long.parseLong(t.split("\\|")[1]);
							return new Tuple2<String, Long>(d_c, count);
						}
						
					} catch (Exception e) {
						if(log.isErrorEnabled()){
							log.error(e.getMessage(),e);
						}
					}
					return new Tuple2<String, Long>("", 0L);
				}
			}).reduceByKey(new Function2<Long, Long, Long>() {
				private static final long serialVersionUID = 1L;
				@Override
				public Long call(Long v1, Long v2) throws Exception {
					return v1+v2;
				}
			});
			
			System.out.println("............将数据放入map，转化成临时表............");
			JavaRDD<DnsCount_jr_city> dnsRDD = visit_count.mapPartitions(new FlatMapFunction<Iterator<Tuple2<String,Long>>, DnsCount_jr_city>() {
				private static final long serialVersionUID = 1L;
				@Override
				public Iterable<DnsCount_jr_city> call(Iterator<Tuple2<String,Long>> it) throws Exception {
					List<DnsCount_jr_city> list = new ArrayList<DnsCount_jr_city>();
					
					while(it.hasNext()){
						Tuple2<String, Long> tuple = it.next();
						String top_domain = tuple._1.toString().split("@")[0];
						String city = tuple._1.toString().split("@")[1];
						if(StringUtils.isNoneBlank(top_domain)){
							DnsCount_jr_city dcc = new DnsCount_jr_city();
							dcc.setTop_domain(top_domain);
							dcc.setCity(city);
							dcc.setCount(tuple._2);
							list.add(dcc);
						}
					}
					return list;
				}
			});
			
			
			DataFrame df_dc = dataSource.sqlContext().createDataFrame(dnsRDD, DnsCount_jr_city.class);
			df_dc.registerTempTable("df_dc_temp");
			
			df_dc.persist(StorageLevel.MEMORY_AND_DISK_SER());
			System.out.println("............topdomain 合并(域名在br2005中)............");
			String joinsql = "select * from ("
					+ "select a.top_domain,a.city,a.count, "
					+ "(CASE WHEN ISNULL(b.license_key) = false THEN B.license_key ELSE  'N/A' END) AS license_key, "
					+ "(CASE WHEN ISNULL(b.site_name) = false THEN b.site_name ELSE  'N/A' END) AS site_name, "
					+ "(CASE WHEN ISNULL(b.PARENT_INDUSTRY_TYPE) = false THEN b.PARENT_INDUSTRY_TYPE ELSE  '221' END) AS PARENT_INDUSTRY_TYPE, "
					+ "ROW_NUMBER() over(partition by a.city order by a.count desc)rank "
					+ "from df_dc_temp a left join br2005_temp b on a.top_domain = b.top_domain "
					+ "group by a.top_domain,a.city,a.count,license_key,site_name,PARENT_INDUSTRY_TYPE)s "
					+ "where s.rank <= 200";
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
            DataFrame oper_tmp = dataSource.sqlContext().sql(joinsql);
            
            String dayStr = commandMap.get("-d");
	        final Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
	        
	        /*JavaRDD<SA7601> result = oper_tmp.toJavaRDD().mapPartitions(new FlatMapFunction<Iterator<Row>, SA7601>() {
				private static final long serialVersionUID = 1L;
				@Override
				public Iterable<SA7601> call(Iterator<Row> t) throws Exception {
					List<SA7601> list_SA7601 = new ArrayList<SA7601>();	
					while(t.hasNext()){
						Row row = t.next();
						SA7601 sa = new SA7601();
						sa.setYear(DateUtil.getCurrentYear(date));
		        		sa.setHalf_year(DateUtil.getHalfYear(date));
		        		sa.setQuarter(DateUtil.getQuarter(date));
		        		sa.setMonth(DateUtil.getCurrentMonth(date));
		        		sa.setWeek(DateUtil.getCurrentWeek(date));
		        		sa.setDay(DateUtil.getCurrentDay(date));
		        		sa.setBuss_type("WEBSITE_RANK_SUM_CITY");
		        		sa.setBuss_value(row.getAs("city").toString());
		        		sa.setRank(Integer.parseInt(row.getAs("rank").toString()));
		        		sa.setDomain(row.getAs("top_domain").toString());
		        	    sa.setWebsite_name(row.getAs("site_name").toString());
		        	    sa.setIcp_num(row.getAs("license_key").toString());
		        	    sa.setTraffic(row.getAs("count").toString());
		        		sa.setSta_range(0);
		        		list_SA7601.add(sa);
					}
					return list_SA7601;
				}
			});*/
	        
	        List<Row> list_row = oper_tmp.toJavaRDD().take(6000);
			List<SA7601> list_SA7601 = new ArrayList<SA7601>();
	        oper_tmp.persist(StorageLevel.MEMORY_AND_DISK_SER());
	        if(list_row.size() > 0){
	        	for(Row row_top_domain : list_row){
	        	    SA7601 sa = new SA7601();
	        		sa.setYear(DateUtil.getCurrentYear(date));
	        		sa.setHalf_year(DateUtil.getHalfYear(date));
	        		sa.setQuarter(DateUtil.getQuarter(date));
	        		sa.setMonth(DateUtil.getCurrentMonth(date));
	        		sa.setWeek(DateUtil.getCurrentWeek(date));
	        		sa.setDay(DateUtil.getCurrentDay(date));
	        		sa.setBuss_type("WEBSITE_RANK_SUM_CITY");
	        		sa.setBuss_value(row_top_domain.getAs("city").toString());
	        		sa.setRank(Integer.parseInt(row_top_domain.getAs("rank").toString()));
	        		sa.setDomain(row_top_domain.getAs("top_domain").toString());
	        	    sa.setWebsite_name(row_top_domain.getAs("site_name").toString());
	        	    sa.setIcp_num(row_top_domain.getAs("license_key").toString());
	        	    sa.setTraffic(row_top_domain.getAs("count").toString());
	        		sa.setSta_range(0);
	        		list_SA7601.add(sa);
	        	}
	        }
	        
	        df_dc.sqlContext().dropTempTable("df_dc_temp");
	        df_dc.unpersist();
	        JavaRDD<SA7601> result = sc.parallelize(list_SA7601);
	    	DataFrame df = dataSource.sqlContext().createDataFrame(result, SA7601.class);
			outlist.add(new OutQueueEntity(outName, df));
			
			CommonUtils.deleteTaskTableData("SA7601", date, "WEBSITE_RANK_SUM_CITY", 0, "day");
		} catch (Exception e) {
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("DomainAnalyzTask_SA7601JR_New:本省接入(地市)exectime: "+min+" min............");
		return outlist;
	}
	

}
